/**
 * Copyright (C) 2020 Baidu, Inc. All Rights Reserved.
 */
package com.baidu.mpks.backendweb.source.service;

import com.baidu.mpks.backendweb.db.dao.StructInfoDbDao;
import com.baidu.mpks.backendweb.db.enums.SourceStatus;
import com.baidu.mpks.backendweb.file.dao.FileInfoDao;
import com.baidu.mpks.backendweb.file.domain.FileAttachment;
import com.baidu.mpks.backendweb.file.domain.FileInfo;
import com.baidu.mpks.backendweb.file.dto.FileAttachmentDto;
import com.baidu.mpks.backendweb.file.dto.FileDataListDto;
import com.baidu.mpks.backendweb.file.service.FileAttachmentService;
import com.baidu.mpks.backendweb.file.service.FileRUDService;
import com.baidu.mpks.backendweb.file.service.FileStorageService;
import com.baidu.mpks.backendweb.file.service.FileUploadDownloadPreviewService;
import com.baidu.mpks.backendweb.ktree.dao.TreeNodeDao;
import com.baidu.mpks.backendweb.ktree.domain.TreeNode;
import com.baidu.mpks.backendweb.ktree.service.TreeService;
import com.baidu.mpks.backendweb.mongo.service.MongoService;
import com.baidu.mpks.backendweb.source.dao.SourceManageDao;
import com.baidu.mpks.backendweb.video.dao.VideoInfoDao;
import com.baidu.mpks.backendweb.video.domain.VideoInfo;
import com.baidu.mpks.common.HttpClientTemplate;
import com.baidu.mpks.common.dto.ListDto;
import com.baidu.mpks.datasource.constant.SourceContentType;
import com.baidu.mpks.datasource.constant.SourceInputType;
import com.baidu.mpks.datasource.dao.SchemaClassDao;
import com.baidu.mpks.datasource.dao.SourceDao;
import com.baidu.mpks.datasource.domain.SchemaClass;
import com.baidu.mpks.datasource.domain.Source;
import com.baidu.mpks.datasource.domain.StructInfo;
import com.baidu.mpks.datasource.service.SourceMappingService;
import com.baidu.mpks.user.dao.UserDao;
import com.baidu.mpks.user.domain.User;
import com.baidu.mpks.util.JacksonUtil;
import com.baidu.mpks.util.MapUtils;
import com.baidu.mpks.util.SecurityUtil;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.core.type.TypeReference;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.mongodb.client.MongoCursor;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.bson.Document;
import org.springframework.aop.framework.AopContext;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Lazy;
import org.springframework.data.domain.Sort;
import org.springframework.data.domain.Sort.Direction;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import javax.annotation.PostConstruct;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.net.URLDecoder;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Date;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Random;
import java.util.Set;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.stream.Collectors;

/**
 * @author cuixuefei
 */
@Service
@Slf4j
public class SourceServiceImpl implements SourceService {

    @Autowired
    SourceManageDao sourceManageDao;

    @Autowired
    SourceDao sourceDao;

    @Autowired
    UserDao userDao;

    @Autowired
    @Lazy
    TreeService treeService;

    @Autowired
    SourceStatisticService sourceStatisticService;

    @Autowired
    SourceMappingService sourceMappingService;

    @Autowired
    private SecurityUtil securityUtil;

    @Value("${spring.data.mongodb.source.collection}")
    String collectionName;

    @Value("${dataSource.url}")
    String dataSourceHost;

    @Value("${dataSource.addApi}")
    String dataSourceAddApi;

    @Value("${dataSource.updateApi}")
    String dataSourceUpdateApi;

    @Value("${dataSource.deleteApi}")
    String dataSourceDeleteApi;

    @Value("${api.host}")
    private String pushHost;

    @Value("${api.pushdata.path}")
    private String pushPath;

    @Value("${api.dataSource.path}")
    private String apiDataSourePath;

    @Value("${api.createtopic.path}")
    private String apiCreateTopicPath;

    @Value("${api.clearSourceCache.path}")
    private String clearSourceCacheApi;

    @Value("${api.deletetopic.path}")
    private String deleteTopicApi;

    @Value("${api.token}")
    private String uploadToken;

    @Value("${message.source.prefix}")
    private String sourcePrefix;

    @Value("${source.thread.size}")
    private Integer threadPoolSize;

    @Autowired
    HttpClientTemplate httpTemplate;

    @Autowired
    SchemaClassDao schemaClassDao;

    @Autowired
    FileInfoDao fileInfoDao;

    @Autowired
    StructInfoDbDao structInfoDbDao;

    @Autowired
    @Lazy
    FileAttachmentService fileAttachmentService;

    @Autowired
    @Lazy
    FileUploadDownloadPreviewService fileUploadDownloadPreviewService;

    @Autowired
    TreeNodeDao treeNodeDao;

    @Autowired
    @Lazy
    FileRUDService fileRUDService;

    @Autowired
    VideoInfoDao videoInfoDao;
    @Autowired
    FileStorageService fileStorageService;

    @Autowired
    MongoService mongoService;

    @Value("${spring.data.mongodb.source.collection}")
    String sourceCollectionName;

    private static ObjectMapper MAPPER = new ObjectMapper();

    @Value("${mongo.find.batchsize}")
    Integer mongoFindBatchsize;

    ThreadPoolExecutor executorService;

    @PostConstruct
    public void threadExecutorInit() {
        executorService = new ThreadPoolExecutor(threadPoolSize, threadPoolSize,
                0L, TimeUnit.MILLISECONDS,
                new LinkedBlockingQueue<>(100));
        executorService.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
    }

    @Override
    public ListDto<Source> searchSource(Integer nodeId, List<String> inputTypes, List<String> contentTypes,
                                        Integer pageSize, Integer pageNumber) {

        Set<Integer> nodeIds = treeService.getChildrensIncludeSelf(nodeId);
        Integer offset = (pageNumber - 1) * pageSize;
        List<Source> sourceList = sourceDao.findByNodeIdAndTypesPage(nodeIds, inputTypes, contentTypes, pageSize,
                offset);
        long total = 0L;
        List<Source> result = new ArrayList<>();
        if (CollectionUtils.isNotEmpty(sourceList)) {
            total = sourceDao.countByNodeIdAndTypesPage(nodeIds, inputTypes, contentTypes);
            // 设置总数
            result = sourceStatisticService.addSourceTotal(sourceList, contentTypes, inputTypes);

            Set<Integer> sourceNodeIds = new HashSet<>();
            Set<String> userIds = new HashSet<>();
            result.forEach(source -> {
                sourceNodeIds.add(source.getNodeId());
                userIds.add(source.getUpdateBy());
            });
            List<TreeNode> treeNodes = treeNodeDao.getNodeByIds(sourceNodeIds);
            List<User> users = userDao.findByUserIds(userIds);
            Map<Integer, TreeNode> treeNodeMap = MapUtils.listToMap(Integer.class, TreeNode.class, "getId", treeNodes);
            Map<String, User> userMap = MapUtils.listToMap(String.class, User.class, "getUserId", users);

            // 设置知识树名称
            for (Source source : result) {
                source.setNodeName(treeNodeMap.get(source.getNodeId()).getName());
                source.setTrueName(userMap.get(source.getUpdateBy()).getTrueName());
            }
        }
        ListDto<Source> listDto = new ListDto<>(result, (int) total, pageSize, pageNumber);
        HashMap<String, Object> results = new HashMap<>();
        results.put("totalSize", total);
        results.put("list", result);
        results.put("pageSize", pageSize);
        results.put("pageNumber", pageNumber);
        return listDto;
    }

    @Override
    public String getToken() {
        return (System.currentTimeMillis() + new Random().nextInt(999999999)) + "";
    }

    @Override
    @Transactional
    public void addSource(Source source) throws IOException {

        source.setCreateBy(securityUtil.getUserIdOrDefault(""));
        source.setUpdateBy(securityUtil.getUserIdOrDefault(""));
        if (source.getInputType().equals(SourceInputType.API_PUSH.getName())) {
            source.setContent(this.getToken());
        }
        if (source.getInputType().equals(SourceInputType.API_PULL.getName())) {
            // 如果是api拉取的话是异步处理，所以状态设置为创建中。其他默认为创建成功
            source.setStatus(Source.TOBUILD);
        }
        sourceManageDao.addSource(source);
        sourceMappingService.setSource(source);

        if (source.getInputType().equals(SourceInputType.API_PUSH.getName())) {
            Integer sourceId = source.getId();
            // 添加所有类目信息，添加到统计表中
            if (source.getContentType().equals(SourceContentType.STRUCT.getName())) {
                List<SchemaClass> schemaAllList = schemaClassDao.getAllClass();
                List<String> className = schemaAllList.stream().map(item -> {
                    JsonNode label = item.getLabel();
                    return JacksonUtil.getString(label, "zh-cn");
                }).collect(Collectors.toList());
                this.saveClass(className, "all", sourceId);
            }
        }
        if (source.getInputType().equals(SourceInputType.API_PUSH.getName()) || source.getInputType().equals(SourceInputType.UPLOAD.getName())) {
            // 创建sourceTopic，根据资源类型创建topic
            Map<String, String> postData = new HashMap<>();
            postData.put("topicName", sourcePrefix + source.getContentType());
            // postData.put("topicName", sourcePrefix + source.getId());
            String topicData = JacksonUtil.toJson(postData);
            log.info("addSource: topicData {}", topicData);
            log.info("addSource: url {}", pushHost + apiCreateTopicPath);
            try {
                String response = httpTemplate.executePostData(pushHost + apiCreateTopicPath, null, topicData);
                JsonNode responseJson = JacksonUtil.toJsonNode(response);
                log.info("addSource: response {}", responseJson);
                if (responseJson.get("errno").asInt() != 0) {
                    this.deleteSource(source);
                    throw new IOException("添加失败");
                }
            } catch (Exception e) {
                this.deleteSource(source);
                throw new IOException("添加失败");
            }
        }

        if (source.getInputType().equals(SourceInputType.API_PULL.getName())) {
            try {
                this.sendApiPullPostData(source, "add");
            } catch (IOException e) {
                sourceManageDao.deleteSource(source.getId());
                throw new RuntimeException(e.getMessage());
            }
        }

    }

    /**
     * 发送数据拉取的请求给下游@zhanglu，创建拉取的资源
     *
     * @param source 资源信息
     * @param action 新增（add）【默认】，更新(update)，删除(delete)
     */
    public void sendApiPullPostData(Source source, String action) throws IOException {
        if (source.getInputType().equals(SourceInputType.API_PULL.getName())) {
            Map<String, Object> pullData = new HashMap<>();
            pullData.put("url", source.getContent());
            pullData.put("id", source.getId().toString());
            pullData.put("nodeId", source.getNodeId().toString());
            pullData.put("accessType", "api-pull");

            String url = dataSourceHost + dataSourceAddApi;
            if (action.equals("update")) {
                url = dataSourceHost + dataSourceUpdateApi;
            }
            if (action.equals("delete")) {
                pullData = new HashMap<>();
                pullData.put("id", source.getId().toString());
                pullData.put("clear", false);
                url = dataSourceHost + dataSourceDeleteApi;
            }

            String postData = MAPPER.writeValueAsString(pullData);
            log.info("sendApiPullPostData: postData {}", postData);
            log.info("sendApiPullPostData: url {}", url);
            String response = httpTemplate.executePostData(url, null, postData);
            log.info("sendApiPullPostData: response {}", response);
        }
    }

    @Override
    @Transactional
    public void updateSource(Source source) throws IOException {
        Source sourceTemp = sourceMappingService.getSource(source.getId());
        if (sourceTemp != null) {
            if (sourceTemp.getInputType().equals(SourceInputType.API_PULL.getName())) {
                sourceManageDao.updateStatusById(source.getId(), Source.TOBUILD);
                structInfoDbDao.deleteBySourceId(source.getId());
                try {
                    this.sendApiPullPostData(source, "update");
                } catch (IOException e) {
                    log.error(e.getMessage(), e);
                    throw new IOException("更新失败");
                }
            }
            sourceManageDao.updateSourceNameAndContent(source.getId(), source.getName(), source.getContent(),
                    securityUtil.getUserIdOrDefault(""));
            sourceMappingService.setSource(source);
        }
    }

    @Override
    @Transactional
    public void deleteSource(Source source) throws IOException {

        source = sourceMappingService.getSource(source.getId());
        if (source != null) {
            if (source.getInputType().equals(SourceInputType.API_PULL.getName())) {
                try {
                    this.sendApiPullPostData(source, "delete");
                } catch (IOException e) {
                    log.error(e.getMessage(), e);
                    throw new IOException("删除失败");
                }
            }

            Integer sourceId = source.getId();
            sourceDao.updateStatusAndErrMsgById(sourceId, SourceStatus.DELETIMG.getCode(), "正在删除，请等待");
            final String contentType = source.getContentType();
            // 删除文件时的data数据全部从mongo中取，只需将@markdel设置为1推送即可，不需要重新组装
            SourceService service = (SourceService) AopContext.currentProxy();
            service.repushDeleteMongoData(contentType, "@sourceId", sourceId, true,
                    sourceMappingService.getSource(sourceId).getContent());
            SourceContentType sourceContentType = SourceContentType.getContentType(contentType);
            switch (sourceContentType) {
                case STRUCT:
                    sourceManageDao.deleteStructInfoBySourceId(sourceId);
                    break;
                case VIDEO:
                    List<VideoInfo> videoInfos = videoInfoDao.listBySourceId(sourceId);
                    for (VideoInfo videoInfo : videoInfos) {
                        executorService.execute(() -> fileStorageService.deleteFile(contentType,
                                videoInfo.getRealurl()));
                    }
                    videoInfoDao.deleteBySourceId(sourceId);
                    break;
                case FILE:
                    List<FileInfo> fileInfos = fileInfoDao.findBySourceId(sourceId);
                    List<Integer> fileIds = fileInfos.stream().map(FileInfo::getId).collect(Collectors.toList());
                    for (FileInfo fileInfo : fileInfos) {
                        executorService.execute(() -> fileRUDService.deleteLocalFile(fileInfo.getPath()));
                    }
                    fileRUDService.deleteRelatedFileInfo(fileIds);
                    fileInfoDao.deleteBySourceId(sourceId);
                    break;

            }

        }
    }

    @Override
    @Async("executor")
    public void repushDeleteMongoData(String contentType, String columnName, Object value,
                                      boolean deleteSource, String token) {

        Criteria criteria = mongoService.buildCriteriaByColumn(columnName, value);
        MongoCursor<Document> cursor = mongoService.getMongoCursorByCriteria(contentType, criteria);
        while (cursor.hasNext()) {
            Document document = cursor.next();
            try {
                Map map = new ObjectMapper().readValue(JacksonUtil.toJson(document),
                        new TypeReference<Map<String, Object>>() {
                        });
                map.remove("_id");
                executorService.execute(() -> {
                    try {
                        map.put("@markdel", 1);
                        if (deleteSource) {
                            pushToSource(map, token);
                        } else {
                            Source source = sourceMappingService.getSource(
                                    Integer.parseInt(map.get("@sourceId").toString()));
                            if (null != source) {
                                String finalToken = source.getContent();
                                pushToSource(map, finalToken);
                            }
                        }
                    } catch (IOException e) {
                        log.error("" + e);
                    }
                });
            } catch (JsonProcessingException e) {
                log.error("" + e);
            }
        }
        if (deleteSource) {
            Source source = sourceMappingService.getSourceByToken(
                    new SourceMappingService.MappingKey(token, contentType));
            sourceManageDao.deleteSource(source.getId());
            sourceMappingService.removeSource(source);
            clearSourceCache(source);
        }
        cursor.close();
    }

    @Override
    public List<User> findUserList(String contentType) {
        List<String> userIds = new ArrayList<>();
        List<User> userList = new ArrayList<>();
        SourceContentType sourceContentType = SourceContentType.getContentType(contentType);
        switch (sourceContentType) {
            case FILE:
                userIds = fileInfoDao.findUserList();
                break;
            case VIDEO:
                userIds = videoInfoDao.findUserList();
                break;
            case STRUCT:
                userIds = structInfoDbDao.findUserList();
                break;
            case WEB:
                // 因为网页的操作人获取的是api推送资源的创建人
                userIds = sourceDao.findUserListByContentType(contentType);
                break;
        }
        if (CollectionUtils.isNotEmpty(userIds)) {
            userList = userDao.findByUserIds(userIds);
        }
        return userList;
    }

    @Override
    public void deleteTopic(String topicName) {
        Map<String, String> param = new HashMap<>();
        param.put("topicName", topicName);
        log.info("deleteTopic: topicName {}", topicName);
        log.info("deleteTopic: url {}", pushHost + deleteTopicApi);
        try {
            String response = httpTemplate.executeGet(pushHost + deleteTopicApi, param);
            JsonNode responseJson = JacksonUtil.toJsonNode(response);
            log.info("deleteTopic: response {}", responseJson);
        } catch (Exception e) {
            log.error("delete topic error, topic name {}, {}", topicName, e);
        }
    }

    @Override
    public void clearSourceCache(Source source) {
        String sourceStr = JacksonUtil.toJson(source);
        log.info("clearSourceCache: source {}", sourceStr);
        log.info("clearSourceCache: url {}", pushHost + clearSourceCacheApi);
        try {
            String response = httpTemplate.executePostData(pushHost + clearSourceCacheApi, null, sourceStr);
            JsonNode responseJson = JacksonUtil.toJsonNode(response);
            log.info("clearSourceCache: response {}", responseJson);
        } catch (Exception e) {
            log.error("clearSourceCache error, source {}, {}", sourceStr, e);
        }
    }

    @Override
    public Long countMongoBySourceId(List<String> contentTypeList, List<Integer> sourceId) {
        Long count = 0L;
        for (String contentType : contentTypeList) {
            Criteria criteria = mongoService.buildCriteriaByColumn( "@sourceId", sourceId);
            Criteria markdelCriteria = mongoService.buildCriteriaByColumn("@markdel", 0);
            count += mongoService.countMongoData(contentType, Map.class, criteria, markdelCriteria);
        }
        return count;
    }

    @Override
    public List<Map> getSturctApiPushDataFromMongoBySourceId(List<Integer> sourceId, long updateTimeL) {
        List<Map> data = new ArrayList<>();
        Criteria sourceIdSearch = null;
        if (updateTimeL == 0) {
            sourceIdSearch = Criteria.where("@sourceId").in(sourceId)
                    .and("@inputType").is(SourceInputType.API_PUSH.getName());
        } else {
            sourceIdSearch = Criteria.where("@sourceId").in(sourceId)
                    .and("@inputType").is(SourceInputType.API_PUSH.getName())
                    .and("@updateTime").gt(updateTimeL).lte(updateTimeL);
        }
        data.addAll(mongoService.findMongoData(SourceContentType.STRUCT, Map.class, null, sourceIdSearch));
        return data;
    }

    @Override
    public List<Map> getFileDataFromMongoBySourceId(List<Integer> sourceId) {
        List<Map> data = new ArrayList<>();
        data.addAll(mongoService.findMongoData(SourceContentType.FILE, Map.class,
                "@sourceId", sourceId, null));
        return data;
    }

    @Override
    public Map<String, Object> getListFromMongo(SearchCriteria sc) {

        Criteria wd = null;
        // 文件类型条件
        Criteria docType = null;
        Criteria updateTime = null;
        Criteria userId = null;
        Set<Integer> nodeIds = treeService.getChildPermittedNode(sc.getNodeId());
        if (CollectionUtils.isEmpty(nodeIds)) {
            return new HashMap<>();
        }
        List<String> inputTypes = sc.getInputTypes();
        List<String> contentTypes = new ArrayList<>();
        contentTypes.add(sc.getContentType());
        // 根据nodeId获取sourceId值
        List<Source> source = sourceDao.findByNodeIdAndTypesPage(nodeIds, inputTypes, contentTypes, null, null);
        List<Integer> sourceIds = source.stream().map(Source::getId).collect(Collectors.toList());
        if (StringUtils.isNotEmpty(sc.getUserId())) {
            if (sc.getContentType().equals(SourceContentType.WEB.getName())) {
                sourceIds = source.stream()
                        .filter(s -> s.getCreateBy().equals(sc.getUserId()))
                        .map(Source::getId)
                        .collect(Collectors.toList());
            }
            if (sc.getContentType().equals(SourceContentType.VIDEO.getName())) {
                userId = Criteria.where("createBy").is(sc.getUserId());
            }
        }
        Criteria sourceIdSearch = Criteria.where("@sourceId").in(sourceIds);
        if (!StringUtils.isEmpty(sc.getWd())) {
            // 检索词
            wd = new Criteria();
            if (sc.getContentType().equals(SourceContentType.FILE.getName())) {
                wd.orOperator(Criteria.where("fileName").regex(sc.getWd()), Criteria.where("tags").regex(sc.getWd()));
            } else if (sc.getContentType().equals(SourceContentType.VIDEO.getName())) {
                wd.orOperator(Criteria.where("title").regex(sc.getWd()), Criteria.where("tags").regex(sc.getWd()),
                        Criteria.where("brief").regex(sc.getWd()));
            } else if (sc.getContentType().equals(SourceContentType.WEB.getName())) {
                wd.orOperator(Criteria.where("title").regex(sc.getWd()));
            }
        }
        if (sc.getBeginTime() != null && sc.getEndTime() != null) {
            // 时间范围
            updateTime = Criteria.where("@updateTime").gt(sc.getBeginTime()).lte(sc.getEndTime());
        }


        // 文件类型 word excel
        if (!CollectionUtils.isEmpty(sc.getDocTypes())) {
            docType = new Criteria();
            if (sc.getContentType().equals(SourceContentType.FILE.getName())) {
                docType = Criteria.where("docType").in(sc.getDocTypes());
            }
        }

        Query query = mongoService.buildQueryByCriteria(sourceIdSearch, wd, updateTime, userId, docType);
        query.fields().exclude("_id").exclude("htmlContent");

        // System.out.println(query);
        // 时间排序
        query.with(new Sort(Direction.DESC, "@updateTime"));

        MongoTemplate mongoTemplate = mongoService.getMongoTemplate(sc.getContentType());
        long count = mongoTemplate.count(query, Map.class, collectionName);
        //
        // Pageable pageable = PageRequest.of(sc.getPageNumber() - 1, sc.getPageSize());
        // query.with(pageable);
        // List<Map> items = mongoTemplate.find(query, Map.class, collectionName);

        long skipRecords = (sc.getPageNumber() - 1) * sc.getPageSize();
        query.skip(skipRecords).limit(sc.getPageSize());
        List<Map> items = mongoTemplate.find(query, Map.class, collectionName);

        List<String> userIds = new ArrayList<>();
        // 文档的id
        List<Integer> fileIds = new ArrayList<>();
        // 资源ID与操作人ID映射 sourceId -> userId
        Map<Integer, String> sourceIdToUpdateByMap = new HashMap<>();
        for (Map<String, Object> item : items) {
            Integer sourceId = Integer.parseInt(item.get("@sourceId").toString());
            Source itemSource = sourceMappingService.getSource(sourceId);
            String sourceName = "";
            String inputType = "";
            String updateBy = "";
            if (itemSource != null) {
                sourceName = itemSource.getName();
                inputType = itemSource.getInputType();
                // 操作人ID
                updateBy = itemSource.getUpdateBy();
            }
            item.put("sourceName", sourceName);
            item.put("inputType", inputType);
            Object createBy = item.get("createBy");
            if (createBy != null) {
                userIds.add(createBy.toString());
            }
            if (updateBy != null) {
                userIds.add(updateBy.toString());
            }
            if (item.get("oriId") != null) {
                fileIds.add(Integer.parseInt(item.get("oriId").toString()));
            }

            sourceIdToUpdateByMap.put(sourceId, updateBy);
        }

        List<User> users = new ArrayList<>();
        if (!CollectionUtils.isEmpty(userIds)) {
            users = userDao.findByUserIds(userIds);
        }
        Map<String, User> idToUserMap = MapUtils.listToMap(String.class, User.class, "getUserId", users);

        Map<Integer, FileInfo> idToFileInfoMap = new HashMap<>();
        Map<Integer, List<FileAttachment>> idToFileAttachments = new HashMap<>();
        if (!CollectionUtils.isEmpty(fileIds)) {
            List<FileInfo> fileInfos = fileInfoDao.findByIds(fileIds);
            idToFileInfoMap = MapUtils.listToMap(Integer.class, FileInfo.class, "getId", fileInfos);
            idToFileAttachments = fileAttachmentService.findAttachmentsByFileIds(fileIds);
        }
        for (Map item : items) {
            String trueName = "";
            // 从mongo数据里获得操作人key
            Object updateBy = item.get("updateBy");
            User user = null;
            if (updateBy != null) {
                user = idToUserMap.get(updateBy.toString());
                if (user != null) {
                    trueName = user.getTrueName();
                }
            } else {
                Integer sourceId = Integer.parseInt(item.get("@sourceId").toString());
                // 从资源与用户的映射里取userId
                String updateById = sourceIdToUpdateByMap.get(sourceId);
                user = idToUserMap.get(updateById);
                if (user != null) {
                    trueName = user.getTrueName();
                }
            }
            item.put("trueName", trueName);
            if (sc.getContentType().equals(SourceContentType.FILE.getName())) {
                if (item.get("oriId") != null) {
                    FileInfo file = idToFileInfoMap.get(Integer.parseInt(item.get("oriId").toString()));
                    List<FileAttachmentDto> fileAttachmentDtos = new ArrayList<>();
                    if (file != null) {
                        item.put("status", file.getStatus());
                        // 如果是文档类型，则需要返回附件信息
                        List<FileAttachment> fileAttachments = idToFileAttachments.get(file.getId());
                        if (!CollectionUtils.isEmpty(fileAttachments)) {
                            fileAttachmentDtos = fileAttachments.stream()
                                    .map(fileAttachment -> FileAttachmentDto.builder().id(fileAttachment.getId())
                                            .fileId(fileAttachment.getFileId()).fileName(fileAttachment.getFileName())
                                            .filePath(fileAttachment.getFilePath()).size(fileAttachment.getSize())
                                            .createTime(fileAttachment.getCreateTime())
                                            .createBy(fileAttachment.getCreateBy()).build())
                                    .collect(Collectors.toList());
                        }
                    }
                    item.put("attachments", fileAttachmentDtos);
                }
            }
            if (sc.getContentType().equals(SourceContentType.VIDEO.getName())) {
                if (item.get("realurl") != null) {
                    String[] urlArrays = item.get("realurl").toString().split("fileName=");
                    if (urlArrays.length == 2) {
                        try {
                            urlArrays[1] = URLDecoder.decode(urlArrays[1], "utf-8");
                        } catch (UnsupportedEncodingException e) {
                            log.error("" + e);
                        }
                        item.put("realurl", String.format("%sfileName=%s", urlArrays[0], urlArrays[1]));
                    }

                }
            }
        }
        // PageImpl<Map> page = (PageImpl<Map>) PageableExecutionUtils.getPage(items, pageable, () -> count);
        HashMap<String, Object> results = new HashMap<>();
        results.put("totalSize", count);
        // results.put("list", page.getContent());
        results.put("list", items);
        results.put("pageSize", sc.getPageSize());
        results.put("pageNumber", sc.getPageNumber());

        return results;
    }

    @Override
    public Map<String, Object> getListFromMySql(SearchCriteria sc, String type) {
        Set<Integer> nodeIds = treeService.getChildPermittedNode(sc.getNodeId());
        if (CollectionUtils.isEmpty(nodeIds)) {
            return new HashMap<>();
        }
        HashMap<String, Object> results = new HashMap<>();
        List<FileDataListDto> fileDtos = new ArrayList<>();
        long count = 0;
        List<String> inputTypes = sc.getInputTypes();
        List<String> contentTypes = new ArrayList<>();
        contentTypes.add(sc.getContentType());
        // 根据nodeId获取sourceId值
        List<Source> source = sourceDao.findByNodeIdAndTypesPage(nodeIds, inputTypes, contentTypes, null, null);
        if (org.apache.commons.collections.CollectionUtils.isNotEmpty(source)) {
            Set<Integer> sourceIds = source.stream().map(Source::getId).collect(Collectors.toSet());
            int offset = (sc.getPageNumber() - 1) * sc.getPageSize();
            Date startTime = null;
            Date endTime = null;
            if (StringUtils.isEmpty(sc.getWd())) {
                sc.setWd(null);
            }
            if (sc.getBeginTime() != null && sc.getEndTime() != null) {
                // 时间范围
                startTime = new Date(sc.getBeginTime());
                endTime = new Date(sc.getEndTime());
            }
            if (StringUtils.isEmpty(sc.getUserId())) {
                sc.setUserId(null);
            }
            count = fileInfoDao.countBySourceAndNameAndType(sourceIds, sc.getWd(), sc.getUserId(), startTime, endTime,
                    sc.getDocTypes());
            if (count > 0) {
                List<FileInfo> fileInfoList = buildFileInfoList(sc, type, sourceIds, offset, startTime, endTime);
                Set<String> userIds = new HashSet<>();
                // 文档的id
                List<Integer> fileIds = new ArrayList<>();
                // 资源ID与操作人ID映射 sourceId -> userId
                Map<Integer, String> sourceIdToUpdateByMap = new HashMap<>();
                Map<Integer, List<FileAttachment>> idToFileAttachments = new HashMap<>();
                for (FileInfo fileInfo : fileInfoList) {
                    Integer sourceId = fileInfo.getSourceId();
                    Source itemSource = sourceMappingService.getSource(sourceId);
                    String sourceName = "";
                    String inputType = "";
                    if (itemSource != null) {
                        sourceName = itemSource.getName();
                        inputType = itemSource.getInputType();
                        if (itemSource.getUpdateBy() != null) {
                            userIds.add(itemSource.getUpdateBy());
                        }
                        fileInfo.setNodeId(itemSource.getNodeId());
                        sourceIdToUpdateByMap.put(sourceId, itemSource.getUpdateBy());
                    }
                    fileInfo.setSourceName(sourceName);
                    fileInfo.setInputType(inputType);
                    String createBy = fileInfo.getCreateBy();
                    String updateBy = fileInfo.getUpdateBy();
                    if (createBy != null) {
                        userIds.add(createBy);
                    }
                    if (updateBy != null) {
                        userIds.add(updateBy);
                    }

                    if (fileInfo.getId() != null) {
                        fileIds.add(fileInfo.getId());
                    }
                }
                idToFileAttachments = fileAttachmentService.findAttachmentsByFileIds(fileIds);
                List<User> users = new ArrayList<>();
                if (!CollectionUtils.isEmpty(userIds)) {
                    users = userDao.findByUserIds(userIds);
                }
                Map<String, User> idToUserMap = new HashMap<>();
                if (!CollectionUtils.isEmpty(users)) {
                    for (User user : users) {
                        // 将用户id放入到map中
                        idToUserMap.put(String.valueOf(user.getUserId()), user);
                    }
                }
                for (FileInfo fileInfo : fileInfoList) {
                    FileDataListDto fileDto = new FileDataListDto();
                    BeanUtils.copyProperties(fileInfo, fileDto);
                    String trueName = "";
                    String updateBy = fileInfo.getUpdateBy();
                    User user = null;
                    if (updateBy != null) {
                        user = idToUserMap.get(updateBy.toString());
                        if (user != null) {
                            trueName = user.getTrueName();
                        }
                    } else {
                        Integer sourceId = fileInfo.getSourceId();
                        // 从资源与用户的映射里取userId
                        String updateById = sourceIdToUpdateByMap.get(sourceId);
                        user = idToUserMap.get(updateById);
                        if (user != null) {
                            trueName = user.getTrueName();
                        }
                    }
                    fileDto.setTrueName(trueName);
                    if (StringUtils.isEmpty(fileInfo.getRemotePath())) {
                        fileDto.setDownloadUrl(fileUploadDownloadPreviewService.downloadUrl(fileInfo.getPath()));
                    } else {
                        fileDto.setDownloadUrl(fileInfo.getRemotePath());
                    }
                    fileDto.setAtDownloadUrl(fileUploadDownloadPreviewService.atDownloadUrl(fileInfo.getParsePath(),
                            fileInfo.getFileName()));
                    if (StringUtils.isEmpty(fileInfo.getUrl())) {
                        fileDto.setUrl(fileUploadDownloadPreviewService.previewUrl(fileDto.getId()));
                    } else {
                        fileDto.setUrl(fileInfo.getUrl());
                    }
                    fileDto.setAtInputType(fileInfo.getInputType());
                    fileDto.setAtNodeId(Arrays.stream(new String[]{fileInfo.getNodeId().toString()}).collect(Collectors.toSet()));
                    fileDto.setAtId(String.format("%d_%s", fileInfo.getSourceId(), fileInfo.getFileNameId()));
                    fileDto.setCreateTimeStamp(fileInfo.getCreateTime().getTime());
                    if (null != fileInfo.getUpdateTime()) {
                        fileDto.setUpdateTimeStamp(fileInfo.getUpdateTime().getTime());
                    }
                    // 如果是文档类型，则需要返回附件信息
                    List<FileAttachment> fileAttachments = idToFileAttachments.get(fileInfo.getId());
                    fileDto.setAttachments(fileAttachments);
                    fileDtos.add(fileDto);
                }
            }
        }
        results.put("totalSize", count);
        results.put("list", fileDtos);
        results.put("pageSize", sc.getPageSize());
        results.put("pageNumber", sc.getPageNumber());

        return results;
    }

    public List<FileInfo> buildFileInfoList(SearchCriteria sc, String type, Set<Integer> sourceIds, int offset,
            Date startTime, Date endTime) {
        List<FileInfo> fileInfoList;
        if ("statistic".equals(type)) {
            fileInfoList = fileInfoDao.findBySourceAndNameAndType(sourceIds, sc.getWd(),
                    sc.getUserId(), startTime, endTime, sc.getDocTypes(), null, null);
        } else {
            fileInfoList = fileInfoDao.findBySourceAndNameAndType(sourceIds, sc.getWd(),
                    sc.getUserId(), startTime, endTime, sc.getDocTypes(), sc.getPageSize(), offset);
        }
        return fileInfoList;
    }

    public Long getStructTypeTotalFromMongo(String name, Integer sourceId) {
        Criteria typeSearch = Criteria.where("@type").is(name);
        Criteria sourceIdSearch = Criteria.where("@sourceId").is(sourceId);
        Criteria markdelSearch = Criteria.where("@markdel").is(0);
        return mongoService
                .countMongoData(SourceContentType.STRUCT.getName(), typeSearch, sourceIdSearch, markdelSearch);
    }

    @Override
    public Map<String, Object> getStructList(SearchCriteria sc) {
        // 定时获取所有类目信息，从Mongo中获取实时的统计数据，并更新struct_info表
        Set<Integer> nodeIds = treeService.getChildPermittedNode(sc.getNodeId());
        if (CollectionUtils.isEmpty(nodeIds)) {
            return new HashMap<>();
        }
        HashMap<String, Object> results = new HashMap<>();
        Integer offset = (sc.getPageNumber() - 1) * sc.getPageSize();
        Date beginTime = null;
        Date endTime = null;
        if (sc.getBeginTime() != null && sc.getEndTime() != null) {
            beginTime = new Date(sc.getBeginTime());
            endTime = new Date(sc.getEndTime());
        }
        if (StringUtils.isEmpty(sc.getUserId())) {
            sc.setUserId(null);
        }
        List<StructInfo> list = sourceDao.findStructInfoByTitleTypeTime(nodeIds, sc.getWd(), sc.getUserId(),
                beginTime, endTime, sc.getPageSize(), offset);
        for (StructInfo item : list) {
            User user = userDao.findByUserId(item.getCreateBy());
            String trueName = "";
            if (user != null) {
                trueName = user.getTrueName();
            }
            item.setTrueName(trueName);
        }
        Integer total = sourceDao.findStructInfoByTitleTypeTimeCount(nodeIds, sc.getWd(), sc.getUserId(),
                beginTime, endTime, sc.getPageSize(), offset);

        results.put("totalSize", total);
        results.put("list", list);
        results.put("pageSize", sc.getPageSize());
        results.put("pageNumber", sc.getPageNumber());

        return results;
    }

    @Override
    @Transactional
    public void deleteStructData(Integer sourceId) {
        sourceManageDao.deleteStructInfoById(sourceId);
    }

    public void addStructInfoWithoutTotal(String name, Integer sourceId) {
        // 在structInfo统计表里面添加一行类目的统计信息
        StructInfo newStructInfo = new StructInfo() {
            {
                setCreateBy(securityUtil.getUserIdOrDefault(""));
                setUpdateBy(securityUtil.getUserIdOrDefault(""));
                setSourceId(sourceId);
                setName(name);
            }
        };

        sourceManageDao.addStructInfoWithoutTotal(newStructInfo);
    }

    @Override
    @Transactional
    public void saveClass(List<String> oriClassName, String action, Integer sourceId) {
        // 主要针对于api推送的数据，将统计字段添加到struct_info表中，否则不好统计api推送的类目对应的数据总数
        // 获取所有api_push的资源source_id，获取增量的类目，添加增量类目
        if (sourceId != null) {
            // 创建新的资源的时候，提供souceId值
            for (String structClass : oriClassName) {
                this.addStructInfoWithoutTotal(structClass, sourceId);
            }
        } else {
            // 创建schemaClass的时候调用全量或者增量的时候
            List<Source> sourceDataList = sourceDao.findAllByInputType(SourceInputType.API_PUSH.getName());
            for (Source source : sourceDataList) {
                log.info("saveClass ------> sourceId {}", source.getId());
                sourceId = source.getId();
                if (action.equals("all")) {
                    for (String structClass : oriClassName) {
                        this.addStructInfoWithoutTotal(structClass, sourceId);
                    }
                }
                if (action.equals("inc")) {
                    List<String> oldClassName = sourceDao.findStructInfoNameBySourceId(sourceId);
                    List<String> diffClassName = new ArrayList<>();
                    diffClassName.addAll(oriClassName);
                    diffClassName.removeAll(oldClassName);
                    log.info("saveClass ------> diffClassName {}", diffClassName);
                    for (String className : diffClassName) {
                        this.addStructInfoWithoutTotal(className, sourceId);
                    }
                }
            }
        }
    }

    @Override
    @Transactional
    public void addStructData(StructInfo structInfo) {
        sourceManageDao.addStructInfoWithTotal(structInfo);
    }

    @Override
    @Transactional
    public void saveStructDataTotalAndStatus(List<StructInfo> structInfoList, Integer sourceId) {
        /*
         * { "sourceId": 1, // 请求时候的id "data": [{ "name": "prop1", // 拉取的数据中类目信息 "total": 0 }], "code": 0/201, // 成功是0
         * "message": "xxx", "traceback": "xxxx" }
         */
        sourceManageDao.updateStatusById(sourceId, Source.SUCCESS);
        for (StructInfo item : structInfoList) {
            String userId = sourceMappingService.getSource(sourceId).getUpdateBy();
            item.setCreateBy(userId);
            item.setUpdateBy(userId);
            item.setSourceId(sourceId);
            sourceManageDao.addStructInfoWithTotal(item);
        }
    }

    @Override
    public void pushToDataSource(Map<String, Object> data) throws IOException {

        String uri = pushHost + apiDataSourePath;
        String postData = new ObjectMapper().writeValueAsString(data);
        log.info("pushToDataSource: uri {}", uri);
        log.info("pushToDataSource: postData id {}", data.get("id"));
        String response = httpTemplate.executePostData(uri, null, postData);
        log.info("pushToDataSource: responseData {}", response);
    }

    @Override
    public String pushToSource(Object data, String token) throws IOException {
        String uri = pushHost + pushPath;
        log.info("pushToSource: url{}", uri);
        String postData = new ObjectMapper().writeValueAsString(data);
        Map<String, String> header = new HashMap<>();
        Map<String, Object> dataMap = (Map) data;
        if (StringUtils.isEmpty(token)) {
            token = uploadToken;
        }
        header.put("token", token);
        header.put("internalData", "true");
        log.info("pushToSource: token {}", token);
        log.info("pushToSource: pushData {}", dataMap.get("@id"));
        String response = httpTemplate.executePostData(uri, null, postData, header);
        log.info("pushToSource: response {}", response);
        return response;
    }

    @Override
    public void sendToDataSource(List<Integer> sourceId, Integer dataSourceId, String dataSourceName,
                                 String contentType, Long startTime, Long endTime) {
        // 获取所有的mongo表中，sourceId相关的资源，推送到datasource的kafka中
        Criteria sourceIdCriteria = Criteria.where("@sourceId").in(sourceId);
        Criteria allData = null;
        if (startTime != null && endTime != null) {
            // 针对于用户上传的文档，用createTime作为接入系统时间；其他类型用推送到source的Kafka中作为接入系统时间
            // 文档数据先写入Mongo生成一个createTime，然后添加到mysql，因为有事务，所以会有一定延迟，mongo中的createTime更新为mysql中的时间。
            Criteria uploadData = Criteria.where("@sourceId").in(sourceId).and("@inputType").is("upload")
                    .and("createTime").gte(startTime).lte(endTime);
            Criteria otherData = Criteria.where("@sourceId").in(sourceId).and("@createTime").gte(startTime)
                    .lte(endTime);
            allData = new Criteria().orOperator(uploadData, otherData);
        }
        MongoCursor<Document> cursor;
        if (allData != null) {
            cursor = mongoService.getMongoCursorByCriteria(contentType, allData);
        } else {
            cursor = mongoService.getMongoCursorByCriteria(contentType, sourceIdCriteria);
        }
        while (cursor.hasNext()) {
            Document document = cursor.next();
            Map map = null;
            try {
                map = new ObjectMapper().readValue(JacksonUtil.toJson(document),
                        new TypeReference<Map<String, Object>>() {
                        });
                map.put("@dataSourceId", dataSourceId);
                map.put("@dataSourceName", dataSourceName);
                log.info("sendDataFromMongoToDataSource: mongoData @id {}", map.get("@id"));
                this.pushToDataSource(map);
            } catch (IOException e) {
                log.error(e.getMessage(), e);
            }
        }
        cursor.close();
    }

    @Override
    @Transactional
    public void deleteByFileIdsOrAtIds(String contentType, List<Integer> fileIds, List<String> atIds) {
        // 删除文件，视频，结构化，网页
        SourceContentType sourceContentType = SourceContentType.getContentType(contentType);
        SourceService service = (SourceService) AopContext.currentProxy();
        switch (sourceContentType) {
            case WEB:
                service.repushDeleteMongoData(contentType, "@id", atIds, false, null);
                break;
            case VIDEO:
                List<Map> idList = mongoService.findMongoData(contentType, Map.class, "@id", atIds, null,
                        new String[]{"@id", "id"});
                service.repushDeleteMongoData(contentType, "@id", atIds, false, null);
                List<Integer> ids =
                        idList.stream().map(map -> Integer.parseInt(map.get("id").toString())).collect(Collectors.toList());
                if (!CollectionUtils.isEmpty(ids)) {
                    List<VideoInfo> videoInfos = videoInfoDao.listByIds(ids);
                    for (VideoInfo videoInfo : videoInfos) {
                        executorService.execute(() -> fileStorageService.deleteFile(contentType,
                                videoInfo.getRealurl()));
                    }
                    videoInfoDao.deleteByIds(ids);
                }
                break;
            case FILE:
                if (!CollectionUtils.isEmpty(fileIds)) {
                    service.repushDeleteMongoData(contentType, "oriId", fileIds, false, null);
                    List<FileInfo> fileInfos = fileInfoDao.findByIds(fileIds);
                    for (FileInfo fileInfo : fileInfos) {
                        executorService.execute(() -> fileRUDService.deleteLocalFile(fileInfo.getPath()));
                    }
                    fileRUDService.deleteRelatedFileInfo(fileIds);
                    fileInfoDao.deleteByIds(fileIds);
                }
                break;
            default:
                log.info("contentType is not file video contentType-->{}", contentType);

        }
    }

}

